package es

import (
	"context"
	"logTransfer/utils"
	"strings"
	"time"

	"github.com/olivere/elastic/v7"
)

var (
	client      *elastic.Client //es客户端
	logDataChan chan *LogData   //日志内容管道
)

//LogData 日志信息
type LogData struct {
	Topic string //kafka的topic
	Data  string //日志内容
}

//Init 初始化连接
func Init(addr string, size, num int) (err error) {
	if !strings.HasPrefix(addr, "http://") {
		addr = "http://" + addr
	}
	//连接
	client, err = elastic.NewClient(elastic.SetURL(addr))
	if err != nil {
		utils.Log.Error("初始化es失败: ", err)
		return
	}
	//初始化管道
	logDataChan = make(chan *LogData, size)
	//执行发送
	for i := 0; i < num; i++ {
		go sendToEs()
	}
	return
}

//SendToChan 将信息存入管道
func SendToChan(data *LogData) {
	logDataChan <- data
}

//sendToEs 发送到es
func sendToEs() {
	for {
		select {
		case msg := <-logDataChan:
			put, err := client.Index().
				Index(msg.Topic).
				BodyJson(msg).
				Do(context.Background())
			if err != nil {
				utils.Log.Errorf("topic=%s的信息发送到es失败: %v\n", msg.Topic, err)
				continue
			}
			utils.Log.Info("发送到es成功, id=", put.Id)
		default:
			time.Sleep(time.Second)
		}
	}
}
